#include <thread>
#include <cerrno>
#include <fmt/format.h>

#include "../coro/epoll_suspended_manager.h"

#include "../util/assert.h"

#include "../sys/task_manager.h"
#include "../sys/fd_guard.h"
#include "../sys/thread_pool.h"

using namespace ::std::chrono_literals;

namespace wilson::coro
{
    epoll_suspend_manager::epoll_suspend_manager(size_t thrcnt)
        : esm_managed_thr_cnt{ thrcnt }
    {
        esm_poller.add(esm_pipe.readfd(), EPOLLIN);
    }

    epoll_suspend_manager::~epoll_suspend_manager()
    {
        // We need awake all `coroutine_handle` not awaked!
        for (size_t i = 0; i < esm_suspended_ctxs.size(); ++i)
        {
            if (auto* cur = esm_suspended_ctxs[i]; !cur->empty())
            {
                if (cur->est_in_coro_handle)
                {
                    auto h = ::std::coroutine_handle<>::from_address(
                        cur->est_in_coro_handle
                    );
                    schedule(*sys::normal_task_mgr::get_from_global(), h);
                }
                if (cur->est_out_coro_handle)
                {
                    auto h = ::std::coroutine_handle<>::from_address(
                        cur->est_out_coro_handle
                    );
                    schedule(*sys::normal_task_mgr::get_from_global(), h);
                }
            }
        }
    }

    static void digest_pipemsg(const sys::fd_guard& fd)
    {
        char dummy_buff[256];
        int rt;
        do
        {
            rt = ::read(fd, &dummy_buff, sizeof(dummy_buff));
            if (rt == -1 && errno == EAGAIN)
                break;
        }
        while (rt);
    }

    void epoll_suspend_manager::
    wait_and_schedule(::std::chrono::milliseconds timeout) noexcept
    {
        ::std::vector<epoll_event> evs;
        esm_poller.wait(
            ::std::back_inserter(evs), 
            ::std::max(0ms, ::std::min(timeout, 3000ms))
        );
        for (auto& item : evs)
        {
            // pipe
            if (item.data.fd == esm_pipe.readfd())
            {
                digest_pipemsg(esm_pipe.readfd());
                esm_poller.mod(esm_pipe.readfd(), EPOLLIN);
            }
            // normal suspended task.
            else
            {
                auto* ctx = esm_suspended_ctxs[item.data.fd];
                auto lock = ctx->get_lock();
                auto [in, out] = ctx->retrive_related_coro_addr(item.events);

                /* If the peer socket been closed/shutdown
                 * then the epoll_wait call return with `EPOLLHUP`
                 * event. Thus, we just awake all associated corotine.
                 */
                if (item.events & (EPOLLHUP | EPOLLERR))
                {
                    item.events |= EPOLLIN | EPOLLOUT;
                }

                /* If there are any events we insterest 
                 * but not occurs this `epoll_wait` call
                 * we just rearm it by call 
                 * `epoll_ctl(fd, events, EPOLL_CTL_MOD, xxx)`.
                 */
                if (auto remain_events = ctx->holded_events(); 
                    remain_events && !(item.events & EPOLLHUP))
                {
                    esm_poller.mod(item.data.fd, ctx->holded_events());
                }

                /* If the peer socket associated with the `item`(for loop item)
                 * has beed closed/shutdown we just delete it
                 * because there are no any remain events we insterest, 
                 * we just awake them all.
                 */
                else esm_poller.del(item.data.fd);
                lock.unlock();

                if ((item.events & EPOLLIN) && in)
                {
                    auto h = ::std::coroutine_handle<>::from_address(in);
                    schedule(*sys::normal_task_mgr::get_from_global(), h);
                }
                if ((item.events & EPOLLOUT) && out)
                {
                    auto h = ::std::coroutine_handle<>::from_address(out);
                    schedule(*sys::normal_task_mgr::get_from_global(), h);
                }
                awake();
            }
        }
    }

    void epoll_suspend_manager::
    add_task(int fd, uint32_t event, ::std::coroutine_handle<> h)
    {
        auto* ctx = esm_suspended_ctxs[fd];
        auto lock = ctx->get_lock();
        bool empty_ctx = ctx->empty();
        ctx->add_event(event, h.address());
        lock.unlock();
        if (empty_ctx) esm_poller.add(fd, event);
        else           esm_poller.mod(fd, event);
    }

    void epoll_suspend_manager::stop()
    {
        awake(2 * esm_managed_thr_cnt);
    }

    void epoll_suspend_manager::awake(size_t cur) noexcept
    {
        for (size_t i = 0; i < cur; ++i)
            ::write(esm_pipe.writefd(), "1", 1);
    }
}
